package com.leesche.netdoll.communication;

import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

import org.apache.commons.codec.binary.Hex;
import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;

import com.leesche.framework.core.concurrent.ThreadUtil;
import com.leesche.framework.core.json.GsonUtil;
import com.leesche.framework.core.lang.ArrayUtil;
import com.leesche.framework.core.lang.StringUtil;
import com.leesche.framework.platform.redis.ListCommand;
import com.leesche.framework.platform.redis.RedisManager;
import com.leesche.netdoll.communication.Settings.RedisModule;
import com.leesche.netdoll.communication.Settings.SenderConfig;
import com.leesche.netdoll.communication.protocol.FrameType;
import com.leesche.netdoll.communication.protocol.encoder.CommandData;
import com.leesche.netdoll.communication.protocol.encoder.CommandManager;

import io.netty.buffer.ByteBuf;
import io.netty.buffer.Unpooled;
import io.netty.channel.Channel;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelFutureListener;

public class OutboundDataSender {

	private static final Logger logger = LogManager.getLogger(OutboundDataSender.class);
	private static final Logger outLogger = LogManager.getLogger("server-outbound-logger");

	private final static ListCommand redis = RedisManager.custom().getListCommand(RedisModule.TERMINAL);

	private volatile static OutboundDataSender instance;

	private final ExecutorService executor;

	private OutboundDataSender() {
		executor = Executors.newFixedThreadPool(SenderConfig.OUTBOUND_THREAD);
	}

	public static OutboundDataSender sender() {
		if (instance == null) {
			synchronized (OutboundDataSender.class) {
				if (instance == null)
					instance = new OutboundDataSender();
			}
		}
		return instance;
	}

	public void start() {
		for (int i = 0; i < SenderConfig.OUTBOUND_THREAD; i++) {
			executor.execute(new SenderTask());
		}
	}

	private final class SenderTask implements Runnable {

		@Override
		public void run() {
			while (!executor.isShutdown()) {
				try {
					String cmdJson = redis.lpop(RedisModule.KEY_OUTBOUND);

					if (StringUtil.isBlank(cmdJson)) {
						ThreadUtil.sleep(10);
						continue;
					}

					logger.info("reading from {} : {}", RedisModule.KEY_OUTBOUND, cmdJson);

					CommandData cd = GsonUtil.toBean(cmdJson, CommandData.class);

					if (cd == null || cd.getCommandId() == null || StringUtil.isBlank(cd.getDeviceId()) || (!CommandManager.newInstance().contains(cd.getCommandId()))) {
						Thread.sleep(10);
						continue;
					}

					Channel channel = Session.session().find(cd.getDeviceId());

					if (channel != null) {
						send(channel, cd);
					}

					ThreadUtil.sleep(1);
				} catch (Exception e) {
					logger.catching(e);
					ThreadUtil.sleep(1);
				}
			}
		}

		private void send(final Channel channel, final CommandData cd) {

			if (!channel.isActive() || cd == null)
				return;

			String deviceId = channel.attr(Session.KEY_DEVICE_ID).get();

			if (StringUtil.isBlank(cd.getGameId())) {
				logger.error("incorrect command cause gameId is empty.");
				return;
			}

			channel.attr(Session.KEY_GAME_ID).set(cd.getGameId());

			byte[] src = CommandManager.newInstance().command(cd.getCommandId(), FrameType.START, Session.session().getSequece(deviceId), cd.getArgs());

			if (ArrayUtil.isEmpty(src))
				return;

			ByteBuf buf = Unpooled.copiedBuffer(src);

			ChannelFuture future = channel.writeAndFlush(buf);

			future.addListener(new ChannelFutureListener() {

				@Override
				public void operationComplete(ChannelFuture f) throws Exception {

					if (!f.isSuccess() && f.cause() != null) {
						logger.catching(f.cause());
					}

					outLogger.info(Logs.getMsg(channel.remoteAddress(), deviceId, Hex.encodeHexString(src)));

				}
			});
		}
	}
}
